package cn.amoqi.springbootrocketmq.v2.controller;

import cn.amoqi.springbootrocketmq.v2.MqHandle;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @Description:
 * @author: zhuiFeng
 * @date: 2021年04月19日 9:57
 */
@RestController
@Slf4j
public class IndexController {

    private final MqHandle mqHandle;

    public IndexController(MqHandle mqHandle) {
        this.mqHandle = mqHandle;
    }

    @RequestMapping
    public void test(String tags, String keys) {
        //例如：tags：222
        boolean wotazi = mqHandle.client().send("wotazi46", tags, keys, "77777777777");
        System.out.println(wotazi);
    }

    /**
     * subExpression：tags组合， ||分割
     */
    @PostConstruct
    public void createConsumer() {
        try {
            DefaultMQPushConsumer consumer = mqHandle.client().createConsumer("group-1", "wotazi46", "111");
            consumer.registerMessageListener(this::consume);
            consumer.start();
        } catch (Exception e) {
            String stackTrace = ExceptionUtils.getStackTrace(e);
            log.error(stackTrace);
        }
    }


    public ConsumeConcurrentlyStatus consume(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            String body = new String(msg.getBody(), StandardCharsets.UTF_8);
            String tags = msg.getTags();
            log.info("获取到消息body:{},tags:{}", body, tags);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

}
